package com.atguigu.edu.realtime.common.util;

import com.alibaba.fastjson.JSONObject;
import com.google.common.base.CaseFormat;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;

public class HBaseUtil {
    //获取连接
    public static Connection getHBaseConnection(){
        try {
            Configuration conf = new Configuration();
            conf.set("hbase.zookeeper.quorum","hadoop104,hadoop105,hadoop106");
          //  conf.set("hbase.zookeeper.quorum","localhost:2181,localhost:2182,localhost:2183");
            Connection connection = ConnectionFactory.createConnection(conf);
            return connection;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    //关闭连接
    public static void closeHBaseConnection(Connection connection){
        if (connection!=null&&!connection.isClosed()){
            try {
                connection.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }
    //在Hbase中建表
    public static void createHBaseTable(Connection hbaseConn,String namespace,String tableName,String ...families){
        if (families.length<1){
            System.out.println("============== you must have one family! ===================");
            return;
        }

        try(Admin admin = hbaseConn.getAdmin()) {
            TableName tableNameObj = TableName.valueOf(namespace, tableName);
            if (admin.tableExists(tableNameObj)){
                System.out.println("============== "+tableName+" has already exit in "+namespace+" ! ==============");
                return;
            }
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);
            for (String family : families) {
                tableDescriptorBuilder.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build());
            }
            admin.createTable(tableDescriptorBuilder.build());
            System.out.println("============== "+tableName+" has been create successfully in "+namespace+" ! ==============");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    //在Hbase中删表
    public static void dropHBaseTable(Connection hbaseConn,String namespace,String tableName){
        try {
            Admin admin = hbaseConn.getAdmin();
            TableName tableNameObj = TableName.valueOf(namespace, tableName);
            if (!admin.tableExists(tableNameObj)){
                System.out.println("============== "+tableName+" doesn't exit in "+namespace+" ! ==============");
                return;
            }else {
                admin.disableTable(tableNameObj);
                admin.deleteTable(tableNameObj);
                System.out.println("============== "+tableName+" has been delete successfully in "+namespace+" ! ==============");
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    //向HBase表中put数据
    public static void putRow(Connection hbaseConn, String namespace, String tableName, String rowKey, String family, JSONObject jsonObject){

        TableName tableNameObj = TableName.valueOf(namespace, tableName);

        try(Table table = hbaseConn.getTable(tableNameObj)) {
            Put put = new Put(Bytes.toBytes(rowKey));

            Set<String> columnNames = jsonObject.keySet();
            for (String columnName : columnNames) {
                String columnValue = jsonObject.getString(columnName);
                if (StringUtils.isNotEmpty(columnValue)){
                    put.addColumn(Bytes.toBytes(family),Bytes.toBytes(columnName),Bytes.toBytes(columnValue));
                }
            }

            table.put(put);
            System.out.println("============== put successfully in "+namespace+"'s "+tableName+"'s "+rowKey+" ! ==============");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //向HBase表中删除数据
    public static void deleteRow(Connection hbaseConn,String namespace,String tableName,String rowKey){
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        try(Table table = hbaseConn.getTable(tableNameObj)) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
            System.out.println("============== delete successfully in "+namespace+"'s "+tableName+"'s "+rowKey+" ! ==============");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //根据rowKey从表中查询数据
    public static <T>T getRow(Connection hbaseConn,String namespace,String tableName,String rowKey,Class<T> clz,boolean... isUnderlineToCamel){
        boolean defaultIsUTC=false;
        if (isUnderlineToCamel.length>0){
            defaultIsUTC=isUnderlineToCamel[0];
        }

        try (Table table = hbaseConn.getTable(TableName.valueOf(namespace, tableName))){
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            List<Cell> cells = result.listCells();
            if (cells!=null && cells.size()>0){
                T obj = clz.newInstance();
                for (Cell cell : cells) {
                    //定义一个对象封装查询的一条数据

                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));

                    if (defaultIsUTC){
                        columnName = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, columnName);
                    }
                    BeanUtils.setProperty(obj,columnName,columnValue);
                }
                return obj;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return null;
    }

    public static void main(String[] args) {
        Connection hBaseConnection = getHBaseConnection();
        System.out.println(getRow(hBaseConnection, "namespace","dim_base_trademark","1", JSONObject.class));
        closeHBaseConnection(hBaseConnection);
    }


    //异步连接Hbase
    public static AsyncConnection getHBaseAsyncConn(){
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum","hadoop104,hadoop105,hadoop106");
        try {

            CompletableFuture<AsyncConnection> asyncConnection = ConnectionFactory.createAsyncConnection(conf);
            AsyncConnection asyncConn = asyncConnection.get();
            return asyncConn;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    //关闭异步连接
    public static void closeHBaseAsyncConn(AsyncConnection asyncConnection){
        if (asyncConnection!=null&&!asyncConnection.isClosed()){
            try {
                asyncConnection.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    //通过异步方法从Hbase查询数据
    public static JSONObject readDimAsync(AsyncConnection asyncConnection,String namespace,String tableName,String rowKey){
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncConnection.getTable(tableNameObj);
        Get get = new Get(Bytes.toBytes(rowKey));
        try {
            Result result = asyncTable.get(get).get();
            List<Cell> cells = result.listCells();
            if (cells!=null && cells.size()>0){
                JSONObject dimJsonO = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    dimJsonO.put(columnName,columnValue);
                }
                return dimJsonO;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    return null;
    }
}
